package com.sh.data.engine.domain.datadev.flink.service.impl;

import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkFunctionDomain;
import com.sh.data.engine.domain.datadev.flink.service.FlinkFunctionService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeConfigService;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkNodeConfigEntity;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Objects;

/**
 * @author：jingchen.hr
 * @descripion:
 * @data:6/16/20221:49 PM
 */
@Service
public class FlinkFunctionServiceImpl implements FlinkFunctionService {

    @Autowired
    private FlinkNodeConfigService flinkNodeConfigService;

    @Override
    public void delete(Long nodeId) {
        flinkNodeConfigService.deleteByNodeId(nodeId);
    }

    @Override
    public FlinkFunctionDomain getActiveByNodeId(Long nodeId) {
        FlinkNodeConfigEntity activeByNodeId = flinkNodeConfigService.getActiveByNodeId(nodeId);
        return Convert.convert(FlinkFunctionDomain.class, activeByNodeId);
    }

    @Override
    public List<FlinkFunctionDomain> getByNodeIds(List<Long> nodeIds) {
        if (CollectionUtils.isEmpty(nodeIds)) {
            return Collections.EMPTY_LIST;
        }

        List<FlinkNodeConfigEntity> byNodeIds = flinkNodeConfigService.getByNodeIds(nodeIds);
        return ConvertUtil.copyProperties(byNodeIds, FlinkFunctionDomain.class);
    }

    @Override
    public void publish(Long nodeId, Integer status, String userId) {
        final FlinkFunctionDomain flinkFunctionDomain = this.getActiveByNodeId(nodeId);
        if (null == flinkFunctionDomain) {
            throw new BusinessException("数据不存在");
        }

        if (Objects.equals(flinkFunctionDomain.getIsPublished(), 1)) {
            throw new BusinessException("已发布，不能重复发布");
        }

        FlinkNodeConfigEntity flinkNodeConfigEntity = new FlinkNodeConfigEntity();
        flinkNodeConfigEntity.setId(flinkFunctionDomain.getId());
        flinkNodeConfigEntity.setIsPublished(1);

        if (Objects.equals(status, 1)) {
            flinkNodeConfigEntity.setPublishTime(new Date());
            flinkNodeConfigEntity.setPublisherId(userId);
        } else {
            flinkNodeConfigEntity.setPublishTime(null);
            flinkNodeConfigEntity.setPublisherId(null);
        }


        UpdateWrapper<FlinkNodeConfigEntity> wrapper = Wrappers.update();
        wrapper.lambda().eq(FlinkNodeConfigEntity::getNodeId, nodeId).eq(FlinkNodeConfigEntity::getIsActive, 1);
        flinkNodeConfigService.update(flinkNodeConfigEntity, wrapper);
    }

    @Override
    public void rollbackVersion(Long nodeId, Long versionId) {

        UpdateWrapper<FlinkNodeConfigEntity> wrapper = Wrappers.update();
        wrapper
            .lambda()
            .eq(FlinkNodeConfigEntity::getNodeId, nodeId)
            .eq(FlinkNodeConfigEntity::getIsActive, 1);
        FlinkNodeConfigEntity flinkNodeConfigEntity = new FlinkNodeConfigEntity();
        flinkNodeConfigEntity.setIsActive(0);
        flinkNodeConfigEntity.setUpdateTime(new Date());
        flinkNodeConfigService.update(flinkNodeConfigEntity, wrapper);
        FlinkNodeConfigEntity entity = new FlinkNodeConfigEntity();
        entity.setIsActive(0);
        entity.setUpdateTime(new Date());

        entity.setId(versionId);
        entity.setIsActive(1);
        entity.setUpdateTime(new Date());
        flinkNodeConfigService.updateById(entity);
    }

    @Override
    public void saveFunction(
        Long nodeId,
        String className,
        Long resourceNodeId,
        String purpose,
        String format,
        String paramDesc,
        String userId,
        Long projectId) {

        LambdaQueryWrapper<FlinkNodeConfigEntity> eq = new LambdaQueryWrapper<FlinkNodeConfigEntity>().eq(
            FlinkNodeConfigEntity::getNodeId, nodeId);
        FlinkNodeConfigEntity flinkFunctionEntity = flinkNodeConfigService.getOne(eq);
        if (null == flinkFunctionEntity) {
            flinkFunctionEntity = new FlinkNodeConfigEntity();
            FlinkNodeConfigEntity.FunctionConfig functionConfig =
                new FlinkNodeConfigEntity.FunctionConfig();
            flinkFunctionEntity.setProjectId(projectId);
            flinkFunctionEntity.setNodeId(nodeId);
            functionConfig.setClassName(className);
            functionConfig.setResourceNodeId(resourceNodeId);
            functionConfig.setPurpose(purpose);
            functionConfig.setFormat(format);
            functionConfig.setParamDesc(paramDesc);
            flinkFunctionEntity.setIsPublished(1);
            flinkFunctionEntity.setCreatorId(userId);
            flinkFunctionEntity.setUpdaterId(userId);
            flinkFunctionEntity.setPublisherId(userId);
            flinkFunctionEntity.setPublishTime(new Date());
            flinkFunctionEntity.setIsActive(1);
            flinkFunctionEntity.setCreateTime(new Date());
            flinkFunctionEntity.setUpdateTime(new Date());
            flinkFunctionEntity.setFunctionConfig(functionConfig);
            flinkNodeConfigService.save(flinkFunctionEntity);
        } else {
            flinkFunctionEntity = new FlinkNodeConfigEntity();
            FlinkNodeConfigEntity.FunctionConfig functionConfig =
                new FlinkNodeConfigEntity.FunctionConfig();
            functionConfig.setClassName(className);
            functionConfig.setResourceNodeId(resourceNodeId);
            functionConfig.setPurpose(purpose);
            functionConfig.setFormat(format);
            functionConfig.setParamDesc(paramDesc);
            flinkFunctionEntity.setUpdaterId(userId);
            flinkFunctionEntity.setUpdateTime(new Date());
            flinkFunctionEntity.setFunctionConfig(functionConfig);
            flinkNodeConfigService.updateById(flinkFunctionEntity);
        }
    }
}
